package cn.doitedu.olap_agg;

import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

public class Job03_VideoPlayAnalysisOlapAgg {

    public static void main(String[] args) {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.enableCheckpointing(5000, CheckpointingMode.EXACTLY_ONCE);
        env.getCheckpointConfig().setCheckpointStorage("file:///d:/ckpt");
        env.setParallelism(1);

        StreamTableEnvironment tenv = StreamTableEnvironment.create(env);

        //
        tenv.executeSql(
                " CREATE TABLE dwd_user_action_kafka (                                   "+
                        " 	session_id STRING,                                            "+
                        " 	event_id STRING,                                              "+
                        " 	action_time BIGINT,                                           "+
                        " 	properties MAP<STRING,STRING>,                                "+
                        "   user_id bigint,                                               "+
                        "   province string,                                              "+
                        " 	city string,                                                  "+
                        " 	region string,                       				          "+
                        " 	video_id as cast(properties['video_id'] as bigint) ,          "+
                        " 	play_id as properties['play_id'],                        	  "+
                        " 	rt as to_timestamp_ltz(action_time,3),                    	  "+
                        " 	watermark for rt as rt                    	                  "+
                        " ) WITH (                                                        "+
                        "     'connector' = 'kafka',                                      "+
                        "     'topic' = 'dwd-user-action',                                "+
                        "     'properties.bootstrap.servers' = 'doitedu:9092',            "+
                        "     'properties.group.id' = 'g008',                             "+
                        "     'scan.startup.mode' = 'latest-offset',                      "+
                        "     'value.format' = 'json',                                    "+
                        "     'value.fields-include' = 'EXCEPT_KEY'                       "+
                        " )				                                                  "
        );


        // 首先，过滤
        tenv.executeSql("create temporary view video_events as " +
                "select * from dwd_user_action_kafka where event_id in ('video_play','video_hb','video_pause','video_resume','video_stop')");



        // 标记分段
        tenv.executeSql(
                        " create temporary view flag_ed as                                                          "+
                        " with tmp as (                                                                             "+
                        " select                                                                                    "+
                        "     session_id                                                                            "+
                        "     ,event_id                                                                             "+
                        "     ,action_time                                                                          "+
                        "     ,user_id                                                                              "+
                        "     ,province                                                                             "+
                        "     ,city                                                                                 "+
                        "     ,region                  		                                                        "+
                        "     ,video_id                                                                             "+
                        "     ,play_id                                                                              "+
                        "     ,rt                   	                                                            "+
                        "     ,if(event_id = 'video_resume',1,0) as flag                                            "+
                        " from video_events )                                                                       "+
                        "                                                                                           "+
                        " select                                                                                    "+
                        "     session_id                                                                            "+
                        "     ,event_id                                                                             "+
                        "     ,action_time                                                                          "+
                        "     ,user_id                                                                              "+
                        "     ,province                                                                             "+
                        "     ,city                                                                                 "+
                        "     ,region                      		                                                    "+
                        "     ,video_id                                                                             "+
                        "     ,play_id                                                                              "+
                        "     ,rt                   	                                                            "+
                        "     ,sum(flag) over(partition by user_id,video_id,play_id order by rt ) as flag           "+
                        " from tmp                                                                                  "
        );


        // 创建doris聚合表的映射表
        tenv.executeSql(
                        " create table doris_sink (                                  "+
                        "      user_id                  bigint                       "+
                        "     ,session_id               string                       "+
                        "     ,video_id                 bigint                       "+
                        "     ,video_title              string                       "+
                        "     ,video_type               string                       "+
                        "     ,video_album              string                       "+
                        "     ,video_author             string                       "+
                        "     ,video_time_amt           bigint                       "+
                        "     ,play_id                  string                       "+
                        "     ,flag                     bigint                       "+
                        "     ,partial_start_time       bigint                       "+
                        "     ,partial_end_time         bigint                       "+
                        " ) WITH (                                                   "+
                        "    'connector' = 'doris',                                  "+
                        "    'fenodes' = 'doitedu:8030',                             "+
                        "    'table.identifier' = 'dws.video_play_analysis_olap',    "+
                        "    'username' = 'root',                                    "+
                        "    'password' = 'root',                                    "+
                        "    'sink.label-prefix' = 'doris_label-017'                 "+
                        " )                                                          "
        );




        tenv.executeSql(
                        " create temporary view agg_ed as  select                         "+
                        "     window_start,                                               "+
                        "     window_end,                                                 "+
                        "     user_id,                                                    "+
                        "     session_id,                                                 "+
                        "     video_id,                                                   "+
                        "     play_id,                                                    "+
                        "     flag,                                                       "+
                        "     proctime() as pt,                                           "+
                        "     min(action_time) as partial_start_time,                     "+
                        "     max(action_time) as partial_end_time                        "+
                        " from table (                                                    "+
                        "     tumble(table flag_ed ,descriptor(rt) ,interval '1' minute)  "+
                        " )                                                               "+
                        " group by                                                        "+
                        "     window_start,                                               "+
                        "     window_end,                                                 "+
                        "     user_id,                                                    "+
                        "     session_id,                                                 "+
                        "     video_id,                                                   "+
                        "     play_id,                                                    "+
                        "     flag                                                        "
        );



        // 创建hbase视频信息维表的映射表
        tenv.executeSql(
                " CREATE TABLE cms_video_hbase (             "+
                        "  id bigint,                        "+
                        "  f ROW<                            "+
                        "       video_name string,           "+
                        "       video_type string,           "+
                        "       video_album string,          "+
                        "       video_author string,         "+
                        " 	    video_timelong bigint            "+
                        " 	>,                                   "+
                        "   primary key(id) not enforced	     "+
                        " ) WITH (                               "+
                        "  'connector' = 'hbase-2.2',            "+
                        "  'table-name' = 'dim_cms_video',       "+
                        "  'zookeeper.quorum' = 'doitedu:2181'   "+
                        " )                                      "
        );


        // 将时长聚合结果 和 视频信息维表关联
        tenv.executeSql(
                        " insert into doris_sink                                  "+
                        " select                                                  "+
                        "     a.user_id,                                          "+
                        "     a.session_id,                                       "+
                        "     a.video_id,                                         "+
                        "     b.video_name as video_title,                        "+
                        "     b.video_type,                                       "+
                        "     b.video_album,                                      "+
                        "     b.video_author,                                     "+
                        "     b.video_timelong as video_time_amt,                 "+
                        "     a.play_id,                                          "+
                        "     a.flag,                                             "+
                        "     a.partial_start_time,                               "+
                        "     a.partial_end_time                                  "+
                        " from agg_ed  a                                          "+
                        " left join cms_video_hbase                               "+
                        " FOR SYSTEM_TIME AS OF a.pt as b on a.video_id = b.id    "
        );





    }

}
